package com.atguigu.wuliu.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.wuliu.app.func.BeanToJsonStrFunction;
import com.atguigu.wuliu.app.func.DimAsyncByForeignKeyFunction;
import com.atguigu.wuliu.app.func.DimAsyncMapFunction;
import com.atguigu.wuliu.bean.DwdTransReceiveDetailBean;
import com.atguigu.wuliu.bean.DwsTransOrgReceiveDayBean;
import com.atguigu.wuliu.utils.DateFormatUtil;
import com.atguigu.wuliu.utils.DorisUtil;
import com.atguigu.wuliu.utils.KafkaUtil;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.WindowAssigner;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;
/*物流域机构粒度揽收当日汇总表*/
public class DwsTransOrgReceiveDay {
    public static void main(String[] args) throws Exception {
        //1.获取环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
       // System.setProperty("HADOOP_USER_NAME","atguigu");
        //2.设置检查点
        env.enableCheckpointing(5000L);
        //3.从kafka读取揽收表数据：tms_dwd_trans_receive_detail
        String topic = "tms_dwd_trans_receive_detail";
        String group = "dws_trans_receive_group";
        DataStreamSource<String> kafakSourceDS = env.fromSource(KafkaUtil.getKafkaSource(topic, group), WatermarkStrategy.noWatermarks(), "kafakSource");
       // kafakSourceDS.print("原始数据");
        //4.数据转为自定义实体类 DwsTransOrgReceiveDayBean
        SingleOutputStreamOperator<DwsTransOrgReceiveDayBean> mapDS = kafakSourceDS.map(
                new MapFunction<String, DwsTransOrgReceiveDayBean>() {
                    @Override
                    public DwsTransOrgReceiveDayBean map(String value) throws Exception {
                        DwdTransReceiveDetailBean dwdBean = JSON.parseObject(value, DwdTransReceiveDetailBean.class);
                        DwsTransOrgReceiveDayBean dwsTransOrgReceiveDayBean = DwsTransOrgReceiveDayBean.builder()
                                .ts(dwdBean.getTs())
                                .cityId(dwdBean.getSenderCityId())
                                .districtId(dwdBean.getSenderDistrictId())
                                .provinceId(dwdBean.getSenderProvinceId())
                                .receiveOrderCountBase(1L)
                                .build();
                        return dwsTransOrgReceiveDayBean;
                    }
                }
        );
        //mapDS.print("转换为实体类");


            //5.关联机构数据：通过地区id关联维度表：dim_base_organ 中的 id，获取机构名：org_name
        SingleOutputStreamOperator<DwsTransOrgReceiveDayBean> joinDS = AsyncDataStream.unorderedWait(
                mapDS,
                new DimAsyncByForeignKeyFunction<DwsTransOrgReceiveDayBean>("dim_base_organ") {
                    @Override
                    public void join(DwsTransOrgReceiveDayBean obj, JSONObject dimInfoJsonObj) {
                        obj.setOrgId(dimInfoJsonObj.getString("id"));
                        obj.setOrgName(dimInfoJsonObj.getString("org_name"));
                    }

                    @Override
                    public Tuple2<String, String> getForeignKey(DwsTransOrgReceiveDayBean obj) {
                        return Tuple2.of("region_id", obj.getDistrictId());
                    }
                }

                , 10, TimeUnit.SECONDS

        );
        //joinDS.print("关联机构：");
        //6.设置watermaerks提取事件事件字段ts
        SingleOutputStreamOperator<DwsTransOrgReceiveDayBean> waterMarksDS = joinDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<DwsTransOrgReceiveDayBean>forBoundedOutOfOrderness(Duration.ZERO)
                        .withTimestampAssigner(
                                TimestampAssignerSupplier.of(
                                        new SerializableTimestampAssigner<DwsTransOrgReceiveDayBean>() {
                                            @Override
                                            public long extractTimestamp(DwsTransOrgReceiveDayBean bean, long recordTimestamp) {
                                                return bean.getTs();
                                            }
                                        }
                                )
                        )
        );
        //7.按照机构、进行分组
        KeyedStream<DwsTransOrgReceiveDayBean, String> keyedDS = waterMarksDS.keyBy(
                new KeySelector<DwsTransOrgReceiveDayBean, String>() {
                    @Override
                    public String getKey(DwsTransOrgReceiveDayBean value) throws Exception {
                        return value.getOrgId();
                    }
                }
        );
        //8.开窗
        WindowedStream<DwsTransOrgReceiveDayBean, String, TimeWindow> withWindowDS = keyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));
        //9.聚合并补充窗口信息
        SingleOutputStreamOperator<DwsTransOrgReceiveDayBean> reduceDS = withWindowDS.reduce(
                new ReduceFunction<DwsTransOrgReceiveDayBean>() {
                    @Override
                    public DwsTransOrgReceiveDayBean reduce(DwsTransOrgReceiveDayBean bean, DwsTransOrgReceiveDayBean lastBean) throws Exception {
                        bean.setReceiveOrderCountBase(bean.getReceiveOrderCountBase() + lastBean.getReceiveOrderCountBase());
                        return bean;
                    }
                },
                //<IN, OUT, KEY, W
                new WindowFunction<DwsTransOrgReceiveDayBean, DwsTransOrgReceiveDayBean, String, TimeWindow>() {
                    @Override
                    public void apply(String key, TimeWindow window, Iterable<DwsTransOrgReceiveDayBean> input, Collector<DwsTransOrgReceiveDayBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String edt = DateFormatUtil.toYmdHms(window.getEnd());
                        String curDate = DateFormatUtil.toPartitionDate(window.getStart());
                        for (DwsTransOrgReceiveDayBean bean : input) {
                            bean.setStt(stt);
                            bean.setEdt(edt);
                            bean.setCurDate(curDate);
                            out.collect(bean);
                        }

                    }
                }
        );
       // reduceDS.print("聚合后数据");
        //补充维度
        SingleOutputStreamOperator<DwsTransOrgReceiveDayBean> withOrgNameStream = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncByForeignKeyFunction<DwsTransOrgReceiveDayBean>("dim_base_organ") {
                    @Override
                    public void join(DwsTransOrgReceiveDayBean bean, JSONObject dimJsonObj) {
                        bean.setOrgName(dimJsonObj.getString("org_name"));
                    }

                    @Override
                    public Tuple2<String, String> getForeignKey(DwsTransOrgReceiveDayBean obj) {
                        return Tuple2.of("id",obj.getOrgId());
                    }


                },
                60, TimeUnit.SECONDS
        );

        // 11.2 补充城市名称
        SingleOutputStreamOperator<DwsTransOrgReceiveDayBean> withCityNameStream = AsyncDataStream.unorderedWait(
                withOrgNameStream,
                new DimAsyncByForeignKeyFunction<DwsTransOrgReceiveDayBean>("dim_base_region_info") {
                    @Override
                    public void join(DwsTransOrgReceiveDayBean bean, JSONObject dimJsonObj) {
                        bean.setCityName(dimJsonObj.getString("name"));
                    }

                    @Override
                    public Tuple2<String, String> getForeignKey(DwsTransOrgReceiveDayBean bean) {
                        return Tuple2.of("id",bean.getCityId());
                    }


                },
                60, TimeUnit.SECONDS
        );

        // 11.3 补充省份名称
        SingleOutputStreamOperator<DwsTransOrgReceiveDayBean> fullStream = AsyncDataStream.unorderedWait(
                withCityNameStream,
                new DimAsyncByForeignKeyFunction<DwsTransOrgReceiveDayBean>("dim_base_region_info") {
                    @Override
                    public void join(DwsTransOrgReceiveDayBean bean, JSONObject dimJsonObj) {
                        bean.setProvinceName(dimJsonObj.getString("name"));
                    }

                    @Override
                    public Tuple2<String,String> getForeignKey(DwsTransOrgReceiveDayBean bean) {
                        return Tuple2.of("id",bean.getProvinceId());
                    }
                },
                60, TimeUnit.SECONDS
        );
        fullStream.print("最终结果");


       // TODO 12. 写出到 doris
        fullStream.map(
                new BeanToJsonStrFunction<>()
        ).sinkTo(DorisUtil.getDorisSink("dws_trans_org_receive_day_base"));



        env.execute();
    }
}
